package com.zf.nospring.rabbitmq.consume_manual_ack_nack_reject;

import com.rabbitmq.client.*;
import com.zf.nospring.rabbitmq.nofair.ConsumerNoFair;
import com.zf.nospring.rabbitmq.utils.RabbitUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 手动确认消息被消费
 * 手动向 rabbitmq 发送消息被消费的确认。
 * 如果消息超过一定时间没有被确认消费，那么消息会重新入队。
 * 如果 消费者通道 或者 连接 没有发送确认消费就断开了，那么那么消息也会重新入队。
 */
public class ConsumerManualAck {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitUtils.getDefaultConnection();
        Channel channel1 = connection.createChannel();
        Channel channel2 = connection.createChannel();

        /**
         * 消费队列，关闭自动确认，这里使用两个通道消费同一个队列，一个通道就是一个消费者，同一条消息只会进入一个通道，可以观察一下 deliveryTag 的变化。
         *
         * 对于多个消费者消费同一个队列，rabbitmq默认采用轮询策略，把消息均匀的分发给每个消费者,是公平分发（公平消费）。
         * 对于如何实现不公平消费，详见{@link ConsumerNoFair}
         */
        channel1.basicConsume("test_ack", false, getDefaultConsumer(channel1));
        channel2.basicConsume("test_ack", false, getDefaultConsumer(channel2));
    }

    /**
     * 创建消费者的消费逻辑
     *
     * @param channel 通道
     */
    public static Consumer getDefaultConsumer(Channel channel) {
        // 创建消费者的消费逻辑
        return new DefaultConsumer(channel) {
            /**
             * 当通道中有从队列中获取到的消息时就会执行该方法处理消息，每次处理通道中的一条消息
             * @param consumerTag 消费者标识：每个消费通道就是一个消费者，他们的 consumerTag 不同；如果同一个通道消费不同的队列，那么就是两个消费者，他们的 consumerTag 也不同
             * @param envelope 消息相关的信息：
             *                  - _exchange：交换机，
             *                  - _routingKey：发送消息时的路由键，
             *                  - _redeliver：是否是重新入队的消息，
             *                  - _deliveryTag：消息的标识，基于通道，每个通道都是从1开始，当前通道每处理一条消息值就+1, 用于消息的消费确认
             * @param properties 消息的属性，如类型、编码、过期时间等
             * @param body 消息本体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 消息的标识，基于通道，每个通道都是从1开始，当前通道每次读取一条消息值就+1, 用于消息的消费确认
                long deliveryTag = envelope.getDeliveryTag();
                // 获取当前通道
                Channel ch = getChannel();
                System.out.printf("channel: %s\n consumerTag：%s，deliveryTag: %d，msg: %s\n", ch.toString(), consumerTag, deliveryTag, new String(body));
                // 手动向 rabbitmq 确认消息已被消费
                /*void basicAck(long deliveryTag, boolean multiple);
                     deliveryTag：用来标识消息的id
                     multiple：是否批量，如果true则将一次性确认当前通道中所有小于等于deliveryTag的且没有被确认的消息，
                                也就是会确认当前通道中在当前消息之前收到的（包含当前）所有没有被确认的消息，不会影响当前消息之后通道中收到的消息。
                                如果false则就只确认当前消息
                 */
                ch.basicAck(deliveryTag, false);
            }
        };
    }
}
